package com.demo.task;

import cn.hutool.json.JSONUtil;
import com.demo.domain.EventEntity;
import com.demo.domain.RealtimeEventEntity;
import com.demo.map.EventProductMapFunction;
import com.demo.top.ActiveProcess;
import com.demo.util.DateUtil;
import com.demo.util.Property;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;

import javax.annotation.Nullable;
import java.util.Properties;

/**
 * 热门商品 -> redis
 *
 * @author XINZE
 */
public class TagProductTask {

    private static final int topSize = 5;

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 开启EventTime
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

		FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder()
				.setHost(Property.getStrValue("redis.host"))
				.setPort(Property.getIntValue("redis.port"))
				.setDatabase(Property.getIntValue("redis.db"))
				.build();

        Properties properties = Property.getKafkaProperties("tagProduct");
        DataStreamSource<String> dataStream = env.addSource(new FlinkKafkaConsumer<String>("event", new SimpleStringSchema(), properties));

        DataStream<RealtimeEventEntity> topProduct = dataStream.map(new EventProductMapFunction())
		         .filter((FilterFunction<EventEntity>) eventEntity -> {
			         // 过滤昨天及之前数据
			         boolean ret = eventEntity.getTime()*1000 > DateUtil.getCurrentDate();
			         if (!ret) {
				         System.out.println("out of date: " + JSONUtil.toJsonStr(eventEntity));
				         return false;
			         }
			         if (eventEntity.getUserId() < -1) {
			         	return false;
			         }
			         return true;
//			         return eventEntity.getTime()*1000 > DateUtil.getCurrentDate();
		         })
				// 自定义周期水印，解决最后一个窗口缺少数据不能触发问题
				.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<EventEntity>() {
					private final long maxDelayTime = 3000;
					@Nullable
					@Override
					public Watermark getCurrentWatermark() {
						return new Watermark(System.currentTimeMillis() - maxDelayTime);
					}

					@Override
					public long extractTimestamp(EventEntity eventEntity, long l) {
						return eventEntity.getTime() * 1000;
					}
				})
                // 抽取时间戳做watermark 以 秒 为单位
//                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<EventEntity>() {
//                    @Override
//                    public long extractAscendingTimestamp(EventEntity logEntity) {
//                        return System.currentTimeMillis();
//                    }
//                })
                // 按照 userId、activeType 按滚动窗口
                .keyBy("userId", "activeType").timeWindow(Time.seconds(30))
//                .aggregate(new CountAggForEvent(), new WindowResultFunctionForEvent())
				.process(new ActiveProcess())
		        ;
//	    topProduct.print();
//        topProduct.addSink(new RealtimeTagRedisSink<>(conf,new RealtimeTagRedisMapper()));

        env.execute("TopN ");
    }
}
